package processfunction;

import bean.SensorReading;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.java.tuple.Tuple;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;


/**
 * KeyedProcessFunction 用来操作 KeyedStream。KeyedProcessFunction 会处理流
 * 的每一个元素，输出为 0 个、1 个或者多个元素。所有的 Process Function 都继承自
 * RichFunction 接口，所以都有 open()、close()和 getRuntimeContext()等方法。而
 * KeyedProcessFunction<K, I, O>还额外提供了两个方法:
 *
 * • processElement(I value, Context ctx, Collector<O> out), 流中的每一个元素都
 * 会调用这个方法，调用结果将会放在 Collector 数据类型中输出。Context 可
 * 以访问元素的时间戳，元素的 key，以及 TimerService 时间服务。Context 还
 * 可以将结果输出到别的流(side outputs)。
 *
 * • onTimer(long timestamp, OnTimerContext ctx, Collector<O> out) 是一个回调
 * 函数。当之前注册的定时器触发时调用。参数 timestamp 为定时器所设定的
 * 触发的时间戳。Collector 为输出结果的集合。OnTimerContext 和
 * processElement 的 Context 参数一样，提供了上下文的一些信息，例如定时器
 * 触发的时间信息(事件时间或者处理时间)
 *
     * Context 和 OnTimerContext 所持有的 TimerService 对象拥有以下方法:
     * • long currentProcessingTime() 返回当前处理时间
     * • long currentWatermark() 返回当前 watermark 的时间戳
     * • void registerProcessingTimeTimer(long timestamp) 会注册当前 key 的
     *   processing time 的定时器。当 processing time 到达定时时间时，触发 timer。
     * • void registerEventTimeTimer(long timestamp) 会注册当前 key的 event time 定
     *   时器。当水位线大于等于定时器注册的时间时，触发定时器执行回调函数。
     * • void deleteProcessingTimeTimer(long timestamp) 删除之前注册处理时间定时
     *   器。如果没有这个时间戳的定时器，则不执行。
     * • void deleteEventTimeTimer(long timestamp) 删除之前注册的事件时间定时
     *   器，如果没有此时间戳的定时器，则不执行。
 */


public class ProcessTest1_KeyedProcessFunction {

    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        DataStreamSource<String> inputStream = env.socketTextStream("192.168.45.132", 7777);

        SingleOutputStreamOperator<SensorReading> dataStream = inputStream.map(line -> {
            String[] fields = line.split(",");
            return new SensorReading(fields[0], new Long(fields[1]), new Double(fields[2]));
        });

        dataStream.keyBy("id")
                .process( new MyProcess() )
                .print();

        env.execute();

    }

    private static class MyProcess extends KeyedProcessFunction<Tuple, SensorReading, Integer> {

        ValueState<Long> tsTimerState;

        @Override
        public void open(Configuration parameters) throws Exception {
            tsTimerState = getRuntimeContext().getState(new ValueStateDescriptor<Long>("ts_timer", Long.class));
        }

        @Override
        public void processElement(SensorReading sensorReading, Context context, Collector<Integer> collector) throws Exception {
            collector.collect(sensorReading.getId().length());

            // context
            context.timestamp();
            context.getCurrentKey();
            context.timerService().currentProcessingTime();
            context.timerService().currentWatermark();
            context.timerService().registerProcessingTimeTimer( context.timerService().currentProcessingTime()+5000L);
            tsTimerState.update(context.timerService().currentProcessingTime()+1000L);
        }

        @Override
        public void onTimer(long timestamp, OnTimerContext ctx, Collector<Integer> out) throws Exception {
            System.out.println(timestamp+"定时器触发");
            ctx.getCurrentKey();
            ctx.timeDomain();
        }

        @Override
        public void close() throws Exception {
            tsTimerState.clear();
        }
    }
}
